package com.zhan.rkforum.event;


import com.alibaba.fastjson.JSONObject;
import com.zhan.rkforum.bean.DiscussPost;
import com.zhan.rkforum.bean.Event;
import com.zhan.rkforum.bean.Message;
import com.zhan.rkforum.service.DiscussPostService;
import com.zhan.rkforum.service.ElasticSearchService;
import com.zhan.rkforum.service.MessageService;
import com.zhan.rkforum.util.CommunityConstant;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Component
@SuppressWarnings("all")
public class EventConsumer implements CommunityConstant {

    private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);

    @Autowired
    private MessageService messageService;


    @Autowired
    private DiscussPostService discussPostService;

    @Autowired
    private ElasticSearchService elasticSearchService;

    @KafkaListener(topics = {TOPIC_COMMENT, TOPIC_LIKE, TOPIC_FOLLOW})
    public void handleMessage(ConsumerRecord<String, Object> record){
        boolean emptyRecord = isEmptyRecord(record);
        if(emptyRecord){
            logger.error("消息内容为空！");
            return;
        }
        String recordString = record.value().toString();
        Event event = JSONObject.parseObject(recordString, Event.class);
        if(ObjectUtils.isEmpty(event)){
            logger.error("消息格式错误！");
            return;
        }

        //向站内发送数据
        Message message = new Message();
        message.setFromId(SYSTEM_USERID);
        message.setToId(Integer.parseInt(event.getEntityUserId()));;
        message.setCreateTime(new Date());
        message.setConversationId(event.getTopic());
        message.setStatus(0);
        //将多余数据装入map中
        HashMap<String, Object> map = new HashMap<>();
        map.put("userId", event.getUserId());
        map.put("entityType", event.getEntityType());
        map.put("entityId", event.getEntityId());
        //将时间的数据也装入map中
        if(event.getData() != null){
            Map<String, Object> datas = event.getData();
            for ( Map.Entry< String, Object> data : datas.entrySet()){
                map.put(data.getKey(), data.getValue());
            }
        }
        //将map设置成content发给站内
        message.setContent(JSONObject.toJSONString(map));
        messageService.addMessage(message);
    }


    /**
     * 消费发帖事件
     */
    @KafkaListener(topics = {TOPIC_PUBLISH})
    public void handlePublishDiscussPost(ConsumerRecord<String, Object> record) throws IOException {
        boolean emptyRecord = isEmptyRecord(record);
        if(emptyRecord){
            logger.error("消息内容为空！");
            return;
        }
        String recordString = record.value().toString();
        Event event = JSONObject.parseObject(recordString, Event.class);
        if(ObjectUtils.isEmpty(event)){
            logger.error("消息格式错误！");
            return;
        }
        int entityId = event.getEntityId();
        DiscussPost discussPost = discussPostService.findDiscussPostById(entityId);
        elasticSearchService.saveDiscussPost(discussPost);
    }

    @KafkaListener(topics = {TOPIC_DELETE})
    public void handleDeleteDiscussPost(ConsumerRecord<String, Object> record) throws IOException {
        boolean emptyRecord = isEmptyRecord(record);
        if(emptyRecord){
            logger.error("消息内容为空！");
            return;
        }
        String recordString = record.value().toString();
        Event event = JSONObject.parseObject(recordString, Event.class);
        if(ObjectUtils.isEmpty(event)){
            logger.error("消息格式错误！");
            return;
        }
        int entityId = event.getEntityId();
        elasticSearchService.deleteDiscussPost(entityId);
    }

    private boolean isEmptyRecord(ConsumerRecord<String, Object> record){
        if(ObjectUtils.isEmpty(record) || ObjectUtils.isEmpty(record.value())){
            logger.error("消息内容为空！");
            return true;
        }
        return false;
    }






}
